/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.lvyh.lightframe.job.admin.schedule;

import com.alibaba.fastjson.JSON;
import com.google.common.collect.Lists;
import com.lvyh.lightframe.job.Constants;
import com.lvyh.lightframe.job.admin.bean.JobTaskDefBean;
import com.lvyh.lightframe.job.admin.common.JobInternalContext;
import com.lvyh.lightframe.job.admin.config.ScheduleConfig;
import com.lvyh.lightframe.job.admin.job.DubboJob;
import com.lvyh.lightframe.job.admin.job.HttpJob;
import com.lvyh.lightframe.job.admin.job.StatefulDubboJob;
import com.lvyh.lightframe.job.admin.job.StatefulHttpJob;
import com.lvyh.lightframe.job.admin.listener.ScheduleServerChangeListener;
import com.lvyh.lightframe.job.admin.manager.ScheduleServerManager;
import com.lvyh.lightframe.job.admin.manager.TaskNodeManager;
import com.lvyh.lightframe.job.admin.processor.JobManageProcessor;
import com.lvyh.lightframe.job.admin.route.ExecuteNodeRouteStrategy;
import com.lvyh.lightframe.job.admin.route.ExecuteNodeRouter;
import com.lvyh.lightframe.job.admin.service.MessageNotifyService;
import com.lvyh.lightframe.job.admin.util.NetUtils;
import com.lvyh.lightframe.job.admin.util.SpringContextUtils;
import com.lvyh.lightframe.job.common.Result;
import com.lvyh.lightframe.job.enums.TaskMode;
import com.lvyh.lightframe.job.enums.TaskType;
import org.apache.commons.lang3.StringUtils;
import org.quartz.CronScheduleBuilder;
import org.quartz.DateBuilder;
import org.quartz.Job;
import org.quartz.JobBuilder;
import org.quartz.JobDataMap;
import org.quartz.JobDetail;
import org.quartz.JobKey;
import org.quartz.Scheduler;
import org.quartz.SchedulerException;
import org.quartz.SimpleScheduleBuilder;
import org.quartz.Trigger;
import org.quartz.TriggerBuilder;
import org.quartz.TriggerKey;
import org.quartz.impl.DirectSchedulerFactory;
import org.quartz.impl.matchers.GroupMatcher;
import org.quartz.simpl.RAMJobStore;
import org.quartz.simpl.SimpleThreadPool;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.util.CollectionUtils;

import javax.annotation.PostConstruct;
import javax.annotation.PreDestroy;
import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;

/**
 * Scheduling manager
 */
@Service
public class ScheduleManager implements ScheduleServerChangeListener {

    private static final Logger logger = LoggerFactory.getLogger(ScheduleManager.class);

    private Scheduler scheduler;

    @Autowired
    private TaskNodeManager taskNodeManager;

    @Autowired
    private MessageNotifyService messageNotifyService;

    @Autowired
    private JobManageProcessor jobManageProcessor;

    @Autowired
    private ScheduleServerManager scheduleServerManager;

    @PostConstruct
    public void init() throws Exception {

        ScheduleConfig scheduleConfig = SpringContextUtils.getBean(ScheduleConfig.class);
        scheduleServerManager.addScheduleServerChangeListener(this);
        SimpleThreadPool simpleThreadPool = new SimpleThreadPool(scheduleConfig.getScheduleThreadCount(), 5);
        simpleThreadPool.setThreadNamePrefix("SimpleThreadPool-thread");
        JobInternalContext.setThreadPool(simpleThreadPool);

        DirectSchedulerFactory.getInstance().createScheduler(Constants.SCHEDULER_NAME, NetUtils.getHostAddress(), simpleThreadPool, new RAMJobStore());
        scheduler = DirectSchedulerFactory.getInstance().getScheduler(Constants.SCHEDULER_NAME);
        scheduler.start();
    }

    public void executeSchedule() throws Exception {
        Set<String> onlineServerNodes = scheduleServerManager.getCurrentServerNodes();
        if (CollectionUtils.isEmpty(onlineServerNodes)) {
            return;
        }

        refreshScheduleJobs(onlineServerNodes);
    }

    public synchronized void refreshScheduleJobs(Set<String> onlineServerNodes) throws Exception {

        List<JobTaskDefBean> allocatedJobList = getLatestAllocatedJob(onlineServerNodes);

        Set<JobKey> schedulingJobs = scheduler.getJobKeys(GroupMatcher.<JobKey>anyGroup());
        //Create a temporary set to store TMP job
        Set<JobKey> tmpJobs = new HashSet<>();
        for (JobKey jobKey : schedulingJobs) {
            if (jobKey.getName().startsWith("job-tmp")) {
                tmpJobs.add(jobKey);
            }
        }
        schedulingJobs.removeAll(tmpJobs);

        // Delete scheduled tasks that no longer belong to the machine or whose task properties (task configuration) have changed
        deleteInvalidJob(schedulingJobs, allocatedJobList);

        // Add a new task that belongs to this machine after route
        addNewJob(allocatedJobList);
    }

    public void addNewJob(List<JobTaskDefBean> allocatedJobList) throws Exception {
        List<JobTaskDefBean> result = Lists.newLinkedList();
        Set<JobKey> schedulingJobs = scheduler.getJobKeys(GroupMatcher.<JobKey>anyGroup());

        for (JobTaskDefBean taskDefBean : allocatedJobList) {
            boolean found = false;
            for (JobKey jobKey : schedulingJobs) {
                if (jobKey.equals(getJobKey(taskDefBean))) {
                    found = true;
                }
            }
            if (!found) {
                result.add(taskDefBean);
            }
        }

        for (JobTaskDefBean taskDefBean : result) {
            addSchedule(taskDefBean, false);
            logger.info("[ScheduleManager] add new schedule job, taskId:{}", taskDefBean.getId());
        }
    }

    public void deleteInvalidJob(Set<JobKey> schedulingJobs, List<JobTaskDefBean> allocatedJobList) throws Exception {
        List<JobKey> willDeleteList = Lists.newLinkedList();
        for (JobKey jobKey : schedulingJobs) {
            JobDetail jobDetail = scheduler.getJobDetail(jobKey);
            Long jobVersion = (Long) jobDetail.getJobDataMap().get("taskVersion");

            boolean found = false;
            for (JobTaskDefBean taskDefBean : allocatedJobList) {
                // Version mismatch indicates that the configuration has changed
                if (getJobKey(taskDefBean).equals(jobKey)) {
                    found = true;
                    if (taskDefBean.getVersionId().longValue() != jobVersion.longValue()) {
                        logger.info("The task [taskid: {}] configuration has changed. The scheduling will be reconfigured,taskDef:{}", taskDefBean.getId(), JSON.toJSONString(taskDefBean));
                        willDeleteList.add(jobKey);
                    }
                }
            }

            //Local unassigned tasks also need to be deleted
            if (!found) {
                willDeleteList.add(jobKey);
            }
        }

        if (!CollectionUtils.isEmpty(willDeleteList)) {
            for (JobKey jobKey : willDeleteList) {
                logger.info("Local stop task scheduling:{}, triggers:{}", jobKey, JSON.toJSONString(scheduler.getTriggersOfJob(jobKey)));
            }
        }
        scheduler.deleteJobs(willDeleteList);
    }

    public void addSchedule(int taskId, boolean isTempExecute) throws Exception {
        JobTaskDefBean taskDef = jobManageProcessor.selectJobTaskDefById(taskId);
        if (taskDef == null) {
            return;
        }

        addSchedule(taskDef, isTempExecute);
    }

    /**
     * Add task to scheduler
     *
     * @param taskDef   taskDef
     * @param isTmpMode Is it temporary
     * @throws SchedulerException
     */
    private void addSchedule(JobTaskDefBean taskDef, boolean isTempExecute) throws SchedulerException {
        if (taskDef == null) {
            logger.error("Failed to add task to scheduler: taskdef is empty");
            return;
        }

        int maxTimeout = JobInternalContext.getScheduleMaxTimeout();
        int maxRetry = JobInternalContext.getScheduleMaxRetryTimes();

        if (taskDef.getExecutorTimeout() == null || taskDef.getExecutorTimeout() < 0 || taskDef.getExecutorTimeout() > maxTimeout) {
            taskDef.setExecutorTimeout(maxTimeout);
        }
        if (taskDef.getExecutorFailRetryCount() == null || taskDef.getExecutorFailRetryCount() < 0) {
            taskDef.setExecutorFailRetryCount(0);
        }
        if (taskDef.getExecutorFailRetryCount() > maxRetry) {
            taskDef.setExecutorFailRetryCount(maxRetry);
        }

        if (!isTempExecute && taskDef.getEndTime() != null && taskDef.getEndTime().getTime() < System.currentTimeMillis()) {
            logger.warn("Abort start task scheduling: the end time has expired,taskId={},stopAt={}", taskDef.getId(), taskDef.getEndTime());
            return;
        }

        JobDataMap dataMap = new JobDataMap();
        dataMap.put("taskVersion", taskDef.getVersionId());
        dataMap.put("taskId", taskDef.getId());
        dataMap.put("name", taskDef.getTaskName());
        dataMap.put("timeout", taskDef.getExecutorTimeout());
        dataMap.put("description", taskDef.getJobDesc());
        dataMap.put("retries", taskDef.getExecutorFailRetryCount());
        dataMap.put("params", taskDef.getExecutorParam());
        dataMap.put("concurrent", taskDef.getConcurrentModel());
        dataMap.put("executorRouteStrategy", taskDef.getExecutorRouteStrategy());

        /**
         * Scheduling mode: 0-system scheduling, 1-manual scheduling
         */
        dataMap.put("mode", isTempExecute ? TaskMode.TMP.getMode() : TaskMode.NORMAL.getMode());

        // According to different task definitions
        Class<? extends Job> taskClass = null;
        if (TaskType.HTTP.getType() == taskDef.getTaskType().intValue()) {
            if (taskDef.getConcurrentModel().intValue() == 1) {
                taskClass = HttpJob.class;
            } else {
                taskClass = StatefulHttpJob.class;
            }

            dataMap.put("httpUrl", taskDef.getHttpUrl());
            dataMap.put("httpMethod", taskDef.getHttpMethod());

        } else if (TaskType.DUBBO.getType() == taskDef.getTaskType().intValue()) {

            if (taskDef.getConcurrentModel().intValue() == 1) {
                taskClass = DubboJob.class;
            } else {
                taskClass = StatefulDubboJob.class;
            }

            dataMap.put("dubboService", taskDef.getDubboService());
            dataMap.put("dubboMethod", taskDef.getDubboMethod());
            dataMap.put("dubboVersion", taskDef.getDubboVersion());
            dataMap.put("dubboGroup", taskDef.getDubboGroup());
        }

        JobKey jobKey = null;
        if (isTempExecute) {
            jobKey = getTmpTaskJobKey(taskDef.getId());
        } else {
            jobKey = getJobKey(taskDef);
        }

        JobDetail jobDetail = JobBuilder.newJob(taskClass).withIdentity(jobKey).setJobData(dataMap).withDescription(taskDef.getJobDesc()).requestRecovery().build();

        Trigger trigger = null;

        if (!isTempExecute) {
            trigger = TriggerBuilder.newTrigger().withIdentity(getTriggerKey(taskDef))
                    .endAt(taskDef.getEndTime()).startAt(taskDef.getStartTime()).forJob(jobDetail)
                    .withSchedule(CronScheduleBuilder.cronSchedule(taskDef.getCronExp()).withMisfireHandlingInstructionDoNothing()).build();

        } else {
            // If it is a temporary task,
            // set the execution time to 10 seconds after the current time without repeated scheduling (different from retry)
            trigger = TriggerBuilder.newTrigger().withIdentity(getTmpTriggerKey(taskDef)).forJob(jobDetail)
                    .withSchedule(SimpleScheduleBuilder.simpleSchedule().withRepeatCount(0))
                    .startAt(DateBuilder.futureDate(10, DateBuilder.IntervalUnit.SECOND)).build();

        }

        if (scheduler.checkExists(jobKey) && isTempExecute) {
            scheduler.deleteJob(jobKey);
        }

        scheduler.scheduleJob(jobDetail, trigger);
        logger.info("Start task scheduling locally:{},dataMap:{}, triggers:{}", getJobKey(taskDef), JSON.toJSONString(dataMap), JSON.toJSONString(trigger));
    }

    public void deleteTmpSchedule(int taskId) {
        try {
            scheduler.deleteJob(getTmpTaskJobKey(taskId));
        } catch (SchedulerException e) {
            throw new RuntimeException(e);
        }
    }

    private JobKey getTmpTaskJobKey(int taskId) {
        return new JobKey("job-tmp-" + taskId);
    }

    private TriggerKey getTmpTriggerKey(JobTaskDefBean taskDef) {
        return new TriggerKey("trigger-tmp-" + taskDef.getId());
    }

    private TriggerKey getTriggerKey(JobTaskDefBean taskDef) {
        return new TriggerKey("trigger-" + taskDef.getId());
    }

    private JobKey getJobKey(JobTaskDefBean def) {
        return new JobKey("job-" + def.getId());
    }

    /**
     * Query the newly assigned job of the local machine
     */
    private List<JobTaskDefBean> getLatestAllocatedJob(Set<String> serverNodes) throws Exception {
        List<JobTaskDefBean> result = Lists.newLinkedList();
        logger.info("Load the latest local task list.");

        List<String> addressList = new ArrayList<>(serverNodes);
        List<JobTaskDefBean> taskDefs = jobManageProcessor.getJobTaskDefList();

        String localIp = NetUtils.getHostAddress();

        for (JobTaskDefBean taskDef : taskDefs) {
            if (StringUtils.equals(taskDef.getScheduleNode(), localIp)) {
                result.add(taskDef);
                continue;
            } else {
                ExecuteNodeRouter executeNodeRouter = ExecuteNodeRouteStrategy.convert(taskDef.getExecutorRouteStrategy()).getRouter();
                Result<String> nodeRouterResult = executeNodeRouter.route(taskDef.getId(), addressList);
                String address = nodeRouterResult.getResult();
                if (address.equals(localIp)) {
                    result.add(taskDef);
                }
            }
        }

        return result;
    }

    @PreDestroy
    public void destroy() throws SchedulerException {
        if (this.scheduler != null) {
            scheduler.shutdown();
        }
    }

    @Override
    public void serverNodeChanged(Set<String> lastestNodes) {
        try {
            logger.info("Latest service node:{}", lastestNodes);
            refreshScheduleJobs(lastestNodes);
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }

    @Override
    public void onInit() {

    }

    @Override
    public void onOffline(Set<String> offlineNodes) {
        try {
            if (this.scheduler != null) {
                scheduler.clear();
            }
        } catch (Exception e) {
            logger.error("Node offline exception", e);
        }

    }

    @Override
    public void onDestroy() {
    }

    @Override
    public void onOnline() {
        try {
            this.executeSchedule();
        } catch (Exception e) {
            logger.error(e.getMessage(), e);
        }
    }
}
